package com.example.demo.delay;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;

import java.util.Properties;

/**
 * <p>
 * 负责启动kafka延迟消费的 ManagerBean 以及 KafkaDelay 的健康检查 bean。
 * 这里需要注意的是两个bean的启动顺序。
 * 这里 KafkaDelayHealthIndicator 依赖了 KafkaDelayConsumerManager 来获取 KafkaDelayConsumer 的整体状态，
 * 所以需要在 kafkaDelayConsumerManager 之后启动。
 * </p>
 *
 * @author liuzaihuan
 * @date 2021/12/9 16:58
 * @desc 负责启动kafka延迟消费的 ManagerBean 以及 KafkaDelay 的健康检查 bean
 */
@Configuration
@EnableConfigurationProperties(KafkaDelayConfig.class)
public class KafkaDelayConsumerBootstrap {

    private Logger log = LoggerFactory.getLogger(KafkaDelayConsumerBootstrap.class);

    @Value("${spring.cloud.stream.kafka.binder.brokers}")
    private String brokersAddress;

    @Bean("kafkaDelayConsumerManager")
    public KafkaDelayConsumerManager kafkaDelayConsumerManager(){
        log.info("kafka delay ... init consumer manager ...");
        return new KafkaDelayConsumerManager();
    }

    @Bean("kafkaDelayHealthIndicator")
    @DependsOn("kafkaDelayConsumerManager")
    public KafkaDelayHealthIndicator kafkaDelayHealthIndicator(){
        log.info("kafka delay ... init health indicator ...");
        return new KafkaDelayHealthIndicator();
    }

    @Bean("kafkaAdminClient")
    public KafkaAdminClient kafkaAdminClient(){
        log.info("kafka delay ... init admin client ...");
        Properties props = new Properties();
        props.put("bootstrap.servers", brokersAddress);
        return (KafkaAdminClient) KafkaAdminClient.create(props);
    }

    @Bean("kafkaDltMessageHandler")
    @DependsOn("kafkaAdminClient")
    public KafkaDltMessageHandler kafkaDltMessageHandler(){
        log.info("kafka delay ... init dlt message handler ...");
        return new KafkaDltMessageHandler();
    }

}
